package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.Cn6BasicDataInfo;
import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.offline.service.AbstractDataCombineService;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Description: 对给定的两个车辆有序点数据进行融合，重排序
 * @author 刘俊
 */
public class TidPointsDataCombineServiceImpl extends AbstractDataCombineService implements Serializable {
    @Override
    public JavaPairRDD<Long, List<Point>> combineByTidAndSortByTime(JavaPairRDD<Long, List<Point>> pointsRdd1, JavaPairRDD<Long, List<Point>> pointsRdd2, Map<Long, Cn6BasicDataInfo> cn6BasicDataInfoMap) {
        return pointsRdd1.leftOuterJoin(pointsRdd2).mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, Tuple2<List<Point>, Optional<List<Point>>>>>, Long, List<Point>>) iter ->
                new Iterator<Tuple2<Long, List<Point>>>() {
                    @Override
                    public boolean hasNext() {
                        return iter.hasNext();
                    }

                    @Override
                    public Tuple2<Long, List<Point>> next() {
                        Tuple2<Long, Tuple2<List<Point>, Optional<List<Point>>>> tuple2 = iter.next();
                        Long tid = tuple2._1();
                        Tuple2<List<Point>, Optional<List<Point>>> pointListTuple2 = tuple2._2();
                        List<Point> points0f37 = pointListTuple2._1();

                        //获取0F37最大的时间点
                        Long maxGpstime = points0f37.stream().max((o1, o2) -> (int) (o1.getGpsTime() - o2.getGpsTime())).map(Point::getGpsTime).orElse(0L);

                        //如果该车不是国六车辆，或者vin为空，不融合，不统计发动机数据
                        List<Point> combinePoints = points0f37;
                        Cn6BasicDataInfo cn6BasicDataInfo = cn6BasicDataInfoMap.get(tid);
                        if (cn6BasicDataInfo != null && StringUtils.isNotBlank(cn6BasicDataInfo.getVin())) {
                            Optional<List<Point>> optionalPointsEngine = pointListTuple2._2();
                            if (optionalPointsEngine.isPresent()) {
                                //过滤掉比0F37时间小的发动机流数据
                                List<Point> points2 = optionalPointsEngine.get().stream().filter(e -> e.getGpsTime() < maxGpstime).collect(Collectors.toList());
                                points0f37.addAll(points2);
                                combinePoints = points0f37.stream().sorted((p1, p2) ->  (int) (p1.getGpsTime() - p2.getGpsTime())).collect(Collectors.toList());
                            }
                        }
                        return new Tuple2<>(tid, combinePoints);
                    }
                });
    }
}
